//
// Created by root on 11/6/18.
//

//#include <stdlib.h>
#include <cstdlib>
#include <stdio.h>
#include <string.h>
#include <dirent.h>
#include <inttypes.h>
#include <stdbool.h>
#include <errno.h>
#include <signal.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <arpa/inet.h>

//#include "utcache.h"
#include "anysocket.h"
#include "ursax.h"
#include "log.h"
#include "debug.h"
#include "netserver.h"
#include "utcache.h"
#include "networking.h"
#include "st-hack.h"
//#include "protocol.h"
//#include "networking.h"

void *g_tmp_data;

extern int log_fd;


int server_loop_terminate(struct NetServer* s)
{
    if (unlikely(!s)) {
        return -1;
    }

    return 0;
}

uint32_t server_get_nconnections(struct NetServer* s)
{
    return s->num_conns;
}

static void netservice_std(void* arg, struct NetServer *ns)
{
    struct OpTableArgs *ops_arg = (struct OpTableArgs *)(ns->data);

    int ret;
    int counter = 0;
    char buf[4096];
    st_netfd_t stfd = (st_netfd_t)arg;
    uint32_t ip;
    uint16_t port;
    get_peer_addr(stfd, &ip, &port);
    st_alloc_rw_lock(stfd, ip, port);
    const int header_size = sizeof(struct ProtocolHeader);
    PUSH_THREAD_ADDR(ip, port);

    while(!ns->stop_all_service) {
        LOG_DEBUG("[%s:%d] enter service routine", __func__, __LINE__);

        ret = st_read_fully(stfd, buf, header_size, 10*1000*1000);
        if (ret < 0) {
            counter++;
            if (ETIME == errno && (counter < NET_SERVICE_TIMEOUT_LIMIT)) {
                LOG_DEBUG("[%s:%d] read protocol header timeout: timeout limit:%d, current:%d",
                          __func__, __LINE__, NET_SERVICE_TIMEOUT_LIMIT, counter);
                continue;
            } else {
                break;
            }
        }
        if (unlikely(ret < header_size)) {
            LOG_DEBUG("%s service: connection closed by remote peer", ops_arg->servername);
            break;
        }
        counter = 0;

        struct ProtocolHeader* header = (struct ProtocolHeader*)buf;
        if (unlikely(header->magic != PROTOCOL_HEADER_MAGIC)) {
            errno = EUNATCH;
            log_socket_error(stfd, "magic number not match");
            break;
        }
        uint16_t op = header->op;
        if (unlikely(op >= ops_arg->nb_ops) || 0 == ops_arg->ops[op].size) {
            LOG_ERROR("invalid operation code op is %d", op);
            break;
        }

        int remain = ops_arg->ops[op].size - header_size;
        if (likely(remain > 0)) {
            ret = st_read_fully(stfd, buf + header_size, remain, URSAX_TIMEOUT);
            if (unlikely(ret < remain)) {
                LOG_ERROR("%s service read got error, "
                          "error op is %d, remain is %d, ret is %d. errno %d: %s",
                          ops_arg->servername, op, remain, ret, errno, strerror(errno));
                break;
            }
        }

        g_tmp_data = ops_arg->data;
        ret = ops_arg->ops[op].handler(stfd, buf);

        if (unlikely(ret < 0)) {
            LOG_DEBUG("%s operation error", ops_arg->servername);
            if (CONNECTION_ERROR(ret))
                break;
        }
        flush_tcp(stfd);
    }

    uint16_t ref_st;
    int count = 100;
    while((ref_st = _get_st_reference(stfd)) != 0){
        LOG_INFO("from addr %s:%d ref is %d != 0. wait, some one is using it",
                str_ip(ip), port, ref_st);
        st_usleep(100*1000);
        if(count-- == 0){
            LOG_WARN("wait so many times and will be quit");
            break;
        }
    }

    POP_THREAD_MAGIC();
    st_free_rw_lock(stfd);
    st_netfd_close(stfd);
    LOG_DEBUG("a %s service exit", ops_arg->servername);

    return;
}

static void* pipe_reader(void* arg)
{
    struct NetServer* s = (struct NetServer*)arg;
    if (s->pipefd_use_to_terminate[0] == 0)
        return NULL;

    char buf[4]; // zym: why 4?
    st_netfd_t stfd = st_netfd_open(s->pipefd_use_to_terminate[0]);
    LOG_INFO("pipe reader before st read");
    ssize_t ret = st_read(stfd, buf, sizeof(buf), ST_UTIME_NO_TIMEOUT);
    int error_number = errno;
    LOG_INFO("pipe reader read return, ret is %d, pid is %d", ret, getpid());
    // st_netfd_close(stfd);	// main thread will do it itself, and it cannot be closed twice

    if (ret < 0) {
        if (likely(error_number == EINTR)) {
            LOG_INFO("pipe_reader aborted by main loop thread");
        }
        else {
            LOG_ERROR("pipe_reader: st_read() ret < 0: %d, %s", error_number, strerror(error_number));
        }
    }
    else {
        LOG_INFO("server_loop_thread is %p", s->server_loop_thread);
        if (s->server_loop_thread)
            st_thread_interrupt(s->server_loop_thread);
    }

    s->pipe_thread = NULL;
    return NULL;
}

static st_thread_t create_pipe_reader(struct NetServer* s)
{
    return st_thread_create(pipe_reader, s, 0, 0);
}

static inline st_netfd_t elastic_accept(st_netfd_t stfd)
{
    st_netfd_t cfd;
    MAKE_FD_ELASTIC(
            cfd = st_accept(stfd, NULL, 0, ST_UTIME_NO_TIMEOUT),
            "netserver st_accept error", NULL);
    return cfd;
}

static void* server_stub(void* arg)
{
    struct NetServer* s = (struct NetServer*)arg;
    s->num_conns++;
//    record_thread();
    s->handler(s->clientfd, s);
//    drecord_thread();
    s->num_conns--;
    LOG_DEBUG("s->num_conns=%d, s->terminated=%d", s->num_conns, s->terminated);
    return NULL;
}

static inline void serve_one_connection(struct NetServer* s)
{
    // if fork in handler, and pipe reader is reading, there will
    // be some problems so kill pipe reader first, and create down
    kill_pipe_reader(s);

    // temporarily clear server_loop_thread, because the thread is lent to handler
    s->server_loop_thread = NULL;

    // do serve
    server_stub(s);
}

static int server_loop(struct NetServer* s)
{
    if(!s->handler) {
        s->handler = &netservice_std;
    }

    int ret = listen(s->serverfd, 50);
    if (unlikely(ret == -1)) {
        LOG_ERROR("netserver listen failed %d: %s", errno, strerror(errno));
        close(s->serverfd);
        return -1;
    }

    st_netfd_t stfd = st_netfd_open_socket(s->serverfd);
    if (unlikely(stfd == NULL)) {
        LOG_ERROR("netserver st_netfd_open_socket failed %d: %s", errno, strerror(errno));
        close(s->serverfd);
        s->serverfd = 0;
        return -1;
    }

    s->pipe_thread = NULL;
    ret = pipe(s->pipefd_use_to_terminate);
    if (unlikely(ret < 0)) {
        LOG_ERROR("failed to create pipe fds");
        s->pipefd_use_to_terminate[0] = s->pipefd_use_to_terminate[1] = 0;
        s->pipe_thread = NULL;
    }
    else {
        s->pipe_thread = create_pipe_reader(s);
        if (NULL == s->pipe_thread) {
            LOG_ERROR("[%s:%d]: failed to create pipe reader, errno: %d, %s",
                      __func__, __LINE__, errno, strerror(errno));
        }
        // st_usleep(1000);
    }

    LOG_INFO("enter server loop");
    s->terminated = false;
    while(!s->terminated) {
        s->server_loop_thread = st_thread_self();
        s->clientfd = elastic_accept(stfd);
        if (unlikely(s->clientfd == NULL)) {
            LOG_INFO("elastic_accept got NULL cfd");
            if (likely(s->terminated)) {
                LOG_INFO("server loop terminated");
            }
            else {
                LOG_ERROR("accept error %d: %s", errno, strerror(errno));
                kill_pipe_reader(s);
            }
            break;
        }

        int fd = st_netfd_fileno(s->clientfd);
        enable_tcp_no_delay(fd);
        enable_tcp_cork(fd);
        set_tcp_rbuf_size(fd);
        set_tcp_wbuf_size(fd);

        LOG_DEBUG("s->num_conns %d", s->num_conns);
        if (likely(s->connection_quota > 0) && unlikely(s->connection_quota-- == 1)) {
            serve_one_connection(s); // zym: why one connection?
            break;
        }

        if (likely(s->concurrent)) {
            st_thread_t new_thread = st_thread_create(server_stub, s, 0, ST_STACK_SIZE);
            if (unlikely(new_thread == NULL)) {
                LOG_ERROR("netserver st_thread_create error %d: %s", errno, strerror(errno));
                st_netfd_close(s->clientfd);
            }
            st_sleep(0); // let sub thread go, so that s->clientfd will not be over written
        }
        else {
            serve_one_connection(s);
            s->pipe_thread = create_pipe_reader(s);
        }
    }

    st_netfd_close(stfd);
    s->serverfd = -1;
    if(s->pipe_thread) {
        st_thread_interrupt(s->pipe_thread);
        s->pipe_thread = NULL;
    }

    while (1) {
        bool wait_conn = (s->num_conns > 0) && (s->wait_on_termination || !s->terminated);

        LOG_INFO("wait for all conns close. num_conns: %d wait_on_termination: "
                 "%d terminated: %d",
                 s->num_conns, s->wait_on_termination, s->terminated);
#define WAIT_USECOND (1000*1000)
        if (wait_conn) {
            st_usleep(WAIT_USECOND);
            continue;
        }
        break;
    }
    s->server_loop_thread = NULL;
    close_terminate_pipefd(s);

    LOG_INFO("[%s:%d] server loop terminated successfully", __func__, __LINE__);
    return 0;
}

int netserver(struct NetServer* s)
{
    if (unlikely(s->handler == NULL)) {
        LOG_INFO("s->handler is NULL, will be set default in function server_loop");
    }

    int osfd = socket(PF_INET, SOCK_STREAM, 0);
    if (unlikely(osfd < 0)) {
        LOG_ERROR("netserver: socket creation failed %d: %s", errno, strerror(errno));
        return osfd;
    }

    int ret = 0;
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(s->port);
    addr.sin_addr.s_addr = INADDR_ANY;
    if (s->bindaddr) {
        if (strcmp(s->bindaddr, "localhost")==0) {
            s->bindaddr = "127.0.0.1";
        }
        in_addr_t x = inet_addr(s->bindaddr);
        if (unlikely(x == INADDR_NONE)) {
            LOG_ERROR("netserver: invalid bind address: %s", s->bindaddr);
            goto error_out;
        }
        addr.sin_addr.s_addr = x;
    }

    ret = bind(osfd, (struct sockaddr *)&addr, sizeof(addr));
    if (unlikely(ret == -1)) {
        LOG_ERROR("netserver: socket bind failed %d: %s", errno, strerror(errno));
        goto error_out;
    }

    s->serverfd = osfd;
    ret = server_loop(s);

    return ret;

error_out:
    close(osfd);
    return -1;
}